package events

import (
	"context"
	"github.com/IBM/sarama"
	"go.uber.org/zap"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/saramax"
	"jk-time/webook/search/domain"
	"jk-time/webook/search/service"
	"time"
)

type ArticleEvent struct {
	Id      int64  `json:"id"`
	Title   string `json:"title"`
	Status  int32  `json:"status"`
	Content string `json:"content"`
}
type ArticleKafkaComsumer struct {
	client  sarama.Client
	l       logger.Logger
	syncSvc service.SyncService
}

func (ArticleEvent) Topic() string {
	return "search_articles"
}
func NewTagKafkaComsumer(client sarama.Client, l logger.Logger, syncSvc service.SyncService) Consumer {
	return &ArticleKafkaComsumer{
		client:  client,
		syncSvc: syncSvc,
		l:       l,
	}
}
func (k *ArticleKafkaComsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("search_article_1", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{ArticleEvent{}.Topic()},
			saramax.NewHandler[ArticleEvent]("search_article_1", k.Consume))
		if err != nil {
			zap.L().Error("退出了消费循环,异常", zap.Error(err))
		}
	}()
	return err
}

func (k *ArticleKafkaComsumer) Consume(msg *sarama.ConsumerMessage, t ArticleEvent) error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	return k.syncSvc.InputArticle(ctx, domain.Article{
		Id:      t.Id,
		Title:   t.Title,
		Status:  t.Status,
		Content: t.Content,
	})
}
